package com.ydl.learning.flink.demo

import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

/**
 * flink wordcount demo
 *
 * @author ydl
 * @since 2020/6/24
 */
object WordCount {

  case class WordWithCount(word: String, count: Long)

  def main(args: Array[String]): Unit = {
    //获取执行器env
    val env = ExecutionEnvironment.getExecutionEnvironment

    //读取文件中的数据
    val text = env.readTextFile("/Volumes/work/code/study/learning-flink/pom.xml")

    //对数据过滤，加工处理
    val counts = text.flatMap {
      //\W 表示[^A-Za-z0-9_]
      _.toLowerCase.split("\\W+").filter {
        _.nonEmpty
      }
    }
      .map {
        (_, 1)
      }
      .groupBy(0)
      .sum(1)

    //action: 触发任务
    counts.print()
    //counts.writeAsText("flink-wc5")
    //如果注释了上面的write而不注释execute，会抛出The last execution refers to the latest call to 'execute()', 'count()', 'collect()', or 'print()'.
    //这是因为对于离线批处理的算子，如：“count()”、“collect()”或“print()”等既有sink功能，还有触发的功能。我们上面调用了print()方法，会自动触发execute，所以最后面的一行执行器没有数据可以执行。
    //env.execute("text WordCount")


  }

}
